/****************************************************************************************
 * 
 *  Author:         CFY
 *  Date:           2025-07-04
 *  Description:    MQTT通信测试
 *  Version:        V1.0
 * 
****************************************************************************************/

#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <MQTTClient.h>

// 定义一个有参宏，出错就会打印错误信息
#define ERROR(x) do{ \
 perror(x); \
 printf("%s-%s-%d\n", __FILE__, __func__, __LINE__); \
}while(0)

#define ADDRESS     "tcp://172.16.197.128:1883"         // MQTT 服务器地址
#define CLIENTID    "ExampleClientSub"                  // MQTT 客户端ID， 唯一
#define PUB_TOPIC    "MQTT_PUB"                         // 发布消息的主题
#define SUB_TOPIC    "MQTT_SUB"                         // 订阅消息的主题
#define PAYLOAD     "Hello World!"                      // 负载，消息体
#define QOS         1                                   // MQTT 模式(0,1,2)
#define TIMEOUT     10000L                              // 操作超时时间



// MQTT消息发布子线程
void* pubPthreadServer(void *arg);

// MQTT消息订阅子线程
void* subPthreadServer(void *arg);

static MQTTClient client;
static MQTTClient_connectOptions conn_opts = MQTTClient_connectOptions_initializer;
static MQTTClient_deliveryToken token;

// 确保变量在多线程环境中正确访问， 记录最后一个确认发送的消息令牌
volatile MQTTClient_deliveryToken deliveredtoken;


// 消息发送确认回调
void delivered(void *context, MQTTClient_deliveryToken dt)
{
    printf("Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}


// 消息接收回调函数， 当订阅的主题有新消息时触发， 负责解析和处理收到的消息
int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    int i;
    char* payloadptr;

    printf("Message arrived\n");
    printf("     topic: %s\n", topicName);
    printf("   message: ");

    payloadptr = message->payload;
    for(i=0; i<message->payloadlen; i++)
    {
        putchar(*payloadptr++);
    }
    putchar('\n');

    MQTTClient_freeMessage(&message);       // 释放消息内存
    MQTTClient_free(topicName);             // 释放主题内存
    return 1;
}

// 连接丢失回调
void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}

int main(int argc, char* argv[])
{
    int rc;

    // 创建 MQTT 客户端实例
    MQTTClient_create(&client, ADDRESS, CLIENTID,
        MQTTCLIENT_PERSISTENCE_NONE, NULL);

    // 配置连接选项
    conn_opts.keepAliveInterval = 20;       // 心跳间隔（秒）
    conn_opts.cleansession = 1;             // 清楚会话标志（1表示每次连接都是新会话）

    // 设置回调函数
    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);

    // 连接到MQTT代理服务器
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    printf("Connected to MQTT Broker at %s\n", ADDRESS);

    // MQTT 订阅服务线程
    pthread_t sub_tid;
    if(pthread_create(&sub_tid, NULL, subPthreadServer, NULL))
    {
        // 创建发送服务线程失败
        ERROR("pthread_create");
        MQTTClient_disconnect(client, TIMEOUT);
        MQTTClient_destroy(&client);
        exit(EXIT_FAILURE);
    }

    // MQTT 发送服务线程
    pthread_t pub_tid;
    if(pthread_create(&pub_tid, NULL, pubPthreadServer, NULL))
    {
        // 创建发送服务线程失败
        ERROR("pthread_create");
        MQTTClient_disconnect(client, TIMEOUT);
        MQTTClient_destroy(&client);
        exit(EXIT_FAILURE);
    }



    // 阻塞等待回收子线程资源
    pthread_join(pub_tid, NULL);
    pthread_join(sub_tid, NULL);
    

    
    // 断开与MQTT代理服务器的连接
    MQTTClient_disconnect(client, 10000);
    // 销毁客户端实例，释放资源
    MQTTClient_destroy(&client);
    return 0;
}

// MQTT消息发布子线程
void* pubPthreadServer(void *arg)
{
    printf("-----MQTT publish server pthread start:[%ld]\n", pthread_self());
    MQTTClient_message pubmsg = MQTTClient_message_initializer;                     // 要发送的 msg 数据包
    // 配置要发布的消息
    pubmsg.payload = PAYLOAD;                                                       // 将消息保存到 pubmsg 结构体中
    pubmsg.payloadlen = (int)strlen(PAYLOAD);                                       // 消息的长度
    pubmsg.qos = QOS;                                                               // 模式
    pubmsg.retained = 0;

    // 异步发送消息
    MQTTClient_publishMessage(client, PUB_TOPIC, &pubmsg, &token);
    printf("publish: [%s]:[%s]\n", PUB_TOPIC, PAYLOAD);

    // 等待消息发送完成
    int rc = MQTTClient_waitForCompletion(client, token, TIMEOUT);
    // 检查发送结果
    if (rc == MQTTCLIENT_SUCCESS) {
        printf("Message with delivery token %d successfully delivered\n", token);
    } else {
        printf("Failed to deliver message, return code %d\n", rc);
    }
}

// MQTT消息订阅子线程
void* subPthreadServer(void *arg)
{
    printf("-----MQTT subscribe server pthread start:[%ld]\n", pthread_self());

    int rc;
    int ch;

    // 订阅消息主题
    MQTTClient_subscribe(client, SUB_TOPIC, QOS);

    do 
    {
        ch = getchar();
    } while(ch!='Q' && ch != 'q');

    // 清理资源
    MQTTClient_unsubscribe(client, SUB_TOPIC);
}